package com.example.rabbitmq;


import com.example.rabbitmq.utils.RabbitMQUtil;
import com.rabbitmq.client.*;
import org.springframework.amqp.core.*;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.annotation.RabbitListeners;
import org.springframework.amqp.rabbit.core.RabbitAdmin;


import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 *  注意：由于测试类的缘故，无法让消费者一直运行监听接受，而且还无法输入,所以不用测试类
 * 而是使用main方法执行
 */
public class Test {

    static ConnectionFactory factory;
    static {
        factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        //factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
    }
    public final String CONNECTION_NAME="testConnection";
    public final String EXCHANGE_NAME="testExchange";
    public final String QUEUE_NAME="testQueue";
    //多个路由键可以映射一个队列。
    public final String ROUTING_KEY="testRoutingKey";

    public static void main(String[] args) throws Exception {
        Test test = new Test();
        test.testPriority();
    }


    /**
     * rabbitMQ快速入门，简单的生产者 —— 消费者
     * 需要理解：
     * 方法：
     * newConnection
     * createChannel
     * queueDelete
     * getMessageCount
     * queueDeclare
     * exchangeDeclare
     * queueBind
     * basicPublish
     * DeliverCallback
     * CancelCallback
     * ConsumerShutdownSignalCallback
     * basicConsume
     * close
     * 参数：
     * 4者关系queue、exchange、routingKey、channelNum还与tcp连接的关系，以及连接与连接工厂的关系
     * consumerTag
     * durable
     * autoAck
     * exclusive
     * autoDelete
     *交换机类型type
     * @throws Exception
     */
    public void quickStart() throws Exception {
        int channelNum=16;
        //#########################################生产者#####################################//
        //获取连接。连接不同，则相互独立
        Connection connection = factory.newConnection();

        //获取连接的信道channel
        Channel channel = connection.createChannel(channelNum);
        /**
         * 在该信道创建一个队列，
         * 参数：
         * queue：创建的队列名
         * durable：如果为ture，则创建一个持久队列(该队列将在服务器重启后存活)，否则是只存放再内存中
         * exclusive：如果为ture，则创建一个独占队列(被限制在此连接中)，即其他连接无法获取该队列,而且当连接关闭后队列即被删除
         * autoDelete：如果为true，则创建一个自动删除队列(服务器将在不再使用时删除它)，即当没有连接使用该队列时，则自动删除
         * arguments：
         */
        AMQP.Queue.DeleteOk deleteOk = channel.queueDelete(QUEUE_NAME, true, true);
        if(deleteOk.getMessageCount()==0)
            //如果声明的队列存在
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //如果声明的交换机，已经存在且类型也一致，则不会报错，也不会覆盖。
        channel.exchangeDeclare(EXCHANGE_NAME,"direct");
        //将queueName队列与交换机进行绑定，绑定的路由键为routingKey
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);
        //通过信道，与消息中间件通信。即发送消息到消息中间件。
        String message="消息";

        //channel.basicPublish(exchangeName,queueName,false,false,null,message.getBytes(StandardCharsets.UTF_8));
        channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY,null,message.getBytes(StandardCharsets.UTF_8));
        //如果没有网络，则会将消息暂存起来，等到有网络的时候再发送到消息中间件，从而确保不会因为网络不通而导致数据丢失
        //#########################################消息中间件#####################################//
        /**
         * 对消息的接收和处理，以及存储。
         */
        //#########################################消费者#####################################//
        DeliverCallback deliverCallback=(consumerTag,publishMessage)->{
            System.out.println("====接受消息回调====");
            System.out.println(new String(publishMessage.getBody(),StandardCharsets.UTF_8));
            System.out.println("该信道该消费者的该消息的唯一id，deliveryTag:"+publishMessage.getEnvelope().getDeliveryTag());
            System.out.println(consumerTag);
        };
        CancelCallback cancelCallback=consumerTag->{
            System.out.println("====取消回调====");
            System.out.println(consumerTag);
        };
        ConsumerShutdownSignalCallback consumerShutdownSignalCallback=(consumerTag,sig)->{
            System.out.println("====关闭回调====");
            System.out.println(sig);
            System.out.println(consumerTag);
        };
        channel.basicConsume(QUEUE_NAME,true,"quickStartTag",deliverCallback,cancelCallback,consumerShutdownSignalCallback);
        TimeUnit.SECONDS.sleep(3);
        channel.close(200,"关闭信道");
        connection.close(200,"关闭连接",1000);
    }




    /**
     * 一个队列有多个消费者，rabbitMQ默认使用轮询工作模式。即消费者a接受一个消息，消费者b接受一个消息，一直轮询
     *需要理解：
     * 轮询工作模式、公平分发
     */
    public void manyConsumersPollingWorkingMode() throws Exception {
        Channel producer = RabbitMQUtil.createChannel(16);
        Channel consumer1 = RabbitMQUtil.createChannel(17);
        Channel consumer2 = RabbitMQUtil.createChannel(18);
        //生产者线程
        new Thread(){
            @Override
            public void run() {
                try {
                    //如果声明的交换机，已经存在且类型也一致，则不会报错，也不会覆盖。
                    producer.exchangeDeclare(EXCHANGE_NAME,"direct");
                    //将queueName队列与交换机进行绑定，绑定的路由键为routingKey
                    producer.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);
                    //通过信道，与消息中间件通信。即发送消息到消息中间件。
                    Scanner in=new Scanner(System.in);
                    while(in.hasNext())
                        producer.basicPublish(EXCHANGE_NAME, ROUTING_KEY,null,in.next().getBytes(StandardCharsets.UTF_8));
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }.start();

        //消费者线程1
        new Thread(){
            @Override
            public void run() {
                try {
                    //消费者接受消息回调
                    DeliverCallback deliverCallback=(consumerTag,publishMessage)->{
                        System.out.println("====消费者1接受消息回调====");
                        System.out.println(new String(publishMessage.getBody(),StandardCharsets.UTF_8));
                        System.out.println("deliveryTag:"+publishMessage.getEnvelope().getDeliveryTag());
                        System.out.println(consumerTag);
                    };

                    CancelCallback cancelCallback=consumerTag->{
                        System.out.println("====消费者1取消回调====");
                        System.out.println(consumerTag);
                    };
                    ConsumerShutdownSignalCallback consumerShutdownSignalCallback=(consumerTag,sig)->{
                        System.out.println("====消费者1关闭回调====");
                        System.out.println(sig);
                        System.out.println(consumerTag);
                    };
                    consumer1.basicConsume(QUEUE_NAME,true,"manyConsumersPollingWorkingMode",deliverCallback,cancelCallback,consumerShutdownSignalCallback);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }.start();

        //消费者线程2
        new Thread(){
            @Override
            public void run() {
                try {
                    //消费者接受消息回调
                    DeliverCallback deliverCallback=(consumerTag, publishMessage)->{
                        System.out.println("====消费者2接受消息回调====");
                        System.out.println(new String(publishMessage.getBody(), StandardCharsets.UTF_8));
                        System.out.println("deliveryTag:"+publishMessage.getEnvelope().getDeliveryTag());
                        System.out.println(consumerTag);

                    };

                    CancelCallback cancelCallback= consumerTag->{
                        System.out.println("====消费者2取消回调====");
                        System.out.println(consumerTag);
                    };
                    ConsumerShutdownSignalCallback consumerShutdownSignalCallback=(consumerTag, sig)->{
                        System.out.println("====消费者2关闭回调====");
                        System.out.println(sig);
                        System.out.println(consumerTag);
                    };
                    consumer2.basicConsume(QUEUE_NAME,true,"manyConsumersPollingWorkingMode",deliverCallback,cancelCallback,consumerShutdownSignalCallback);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }.start();
    }


    /**
     * 模拟消息的应答机制
     * 当消费者宕机或者断开连接时，会将消息转发给其他消费者消费，这时候消息唯一标识是新消费者的，前提是该消息还没有被回应确认，一旦被确认就会被删除。
     * 需要理解：
     * 消息应答、批量应答、消息的唯一标识、消息取消
     */
    void messageResponseMechanism() throws Exception {


        Channel producer = RabbitMQUtil.createChannel(16);
        Channel consumer1 = RabbitMQUtil.createChannel(17);
        Channel consumer2 = RabbitMQUtil.createChannel(18);


        //如果声明的交换机，已经存在且类型也一致，则不会报错，也不会覆盖。
        producer.exchangeDeclare(EXCHANGE_NAME,"direct");
        producer.queueDeclare(QUEUE_NAME,false,false,false,null);
        //将queueName队列与交换机进行绑定，绑定的路由键为routingKey
        producer.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);
        //创建一个定时线程池
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
        /**
         * 作用：周期性的执行给定的任务
         * 参数：
         * command – 要执行的任务
         * initialDelay – 延迟第一次执行的时间
         * period – 周期性执行的时间间隔
         * unit – initialDelay和period参数的时间单位
         */
        //生产者线程的周期性任务
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            Date now = new Date(System.currentTimeMillis());
            //每隔20秒发一次消息
            try {
                System.out.println("=======================================================");
                producer.basicPublish(EXCHANGE_NAME, ROUTING_KEY,null,now.toString().getBytes(StandardCharsets.UTF_8));
            } catch (IOException e) {
                e.printStackTrace();
            }
        },0,13,TimeUnit.SECONDS);

        //消费者1线程
        ExecutorService executorService = Executors.newCachedThreadPool();
        executorService.execute(() -> {
            try {
                //消费者接受消息回调
                DeliverCallback deliverCallback=(consumerTag,publishMessage)->{
                    System.out.println("消费者1接受到"+new String(publishMessage.getBody(),StandardCharsets.UTF_8)+"，正在处理中，还未进行消息确认回应");
                    if(Math.ceil(Math.random()*4)==1){
                        try {
                            /**
                             *
                             */
                            consumer1.close(200,"消费者1信道关闭");
                            System.out.println("********************************************");
                            System.out.println("消费者1信道关闭");
                        } catch (TimeoutException e) {
                            e.printStackTrace();
                        }
                    }else{
                        System.out.println("====消费者1接受消息回调====");
                        System.out.println(new String(publishMessage.getBody(),StandardCharsets.UTF_8));
                        System.out.println("该信道该消费者的该消息的唯一标识deliveryTag:"+publishMessage.getEnvelope().getDeliveryTag());
                        System.out.println(consumerTag);
                        System.out.println("消费者1：消息处理完毕，进行消息确认回应");
                        consumer1.basicAck(publishMessage.getEnvelope().getDeliveryTag(),false);
                    }
                };
                //当队列被删除时，回调。队列不存在，并不会回调
                CancelCallback cancelCallback=consumerTag->{
                    System.out.println("队列被删除，则会回调");
                    System.out.println("====消费者1取消回调====");
                    System.out.println(consumerTag);
                };
                ConsumerShutdownSignalCallback consumerShutdownSignalCallback=(consumerTag,sig)->{
                    System.out.println("====消费者1关闭回调====");
                    System.out.println(sig);
                    System.out.println(consumerTag);
                };
                //设置取消自动确认。消费者如果队列不存在，则刚启动消费会报错
                consumer1.basicConsume(QUEUE_NAME, false,"messageResponseMechanism",deliverCallback,cancelCallback,consumerShutdownSignalCallback);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        //消费者2线程
        executorService.execute(() -> {
            try {
                //消费者接受消息回调
                DeliverCallback deliverCallback=(consumerTag, publishMessage)->{
                    System.out.println("消费者2接受到"+new String(publishMessage.getBody(), StandardCharsets.UTF_8)+"，正在处理中，还未进行消息确认回应");
                    System.out.println("====消费者2接受消息回调====");
                    System.out.println(new String(publishMessage.getBody(), StandardCharsets.UTF_8));
                    System.out.println("该信道该消费者的该消息的唯一标识deliveryTag:"+publishMessage.getEnvelope().getDeliveryTag());
                    System.out.println(consumerTag);
                    System.out.println("消费者2：消息处理完毕，进行消息确认回应");
                    /**
                     * 参数：
                     *deliveryTag – 向消息中间件（AMQP实体）发送消息状态，消息确认：  AMQP.Basic.GetOk 或者 AMQP.Basic.Deliver
                     * multiple – 是否批量确认，true代表批量确认
                     */
                    consumer2.basicAck(publishMessage.getEnvelope().getDeliveryTag(),false);
                };
                CancelCallback cancelCallback= consumerTag->{
                    System.out.println("队列被删除，则会回调");
                    System.out.println("====消费者2取消回调====");
                    System.out.println(consumerTag);
                };
                ConsumerShutdownSignalCallback consumerShutdownSignalCallback=(consumerTag, sig)->{
                    System.out.println("====消费者2关闭回调====");
                    System.out.println(sig);
                    System.out.println(consumerTag);
                };
                //设置取消自动确认
                consumer2.basicConsume(QUEUE_NAME, false,"messageResponseMechanism",deliverCallback,cancelCallback,consumerShutdownSignalCallback);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }
    /**
     * 不公平分发机制
     * 需要理解：
     * 信道深入理解：信道可以有多个消费者
     * basicQos
     * 参数：
     * prefetchSize:限制消息的最大字节，如果无限制则为0，默认为0
     * prefetchCount：设置信道最多可以存放prefetchCount条未确认数据（如果达到该值，则就不会再接收信息，而是转发给其他消费者处理）
     * Global：如果为true，代表该信道上所有消费者都会被设置????，不知道怎么设置信道上有多个消费者
     */
    void unfairDeliver() throws Exception {
        Channel producer = RabbitMQUtil.createChannel(16);
        Channel consumer1 = RabbitMQUtil.createChannel(17);
        Channel consumer2 = RabbitMQUtil.createChannel(18);

        /**
         * prefetchSize:限制消息的最大字节，如果无限制则为0，默认为0
         * prefetchCount：设置信道最多可以存放prefetchCount条未确认数据（如果达到该值，则就不会再接收信息，而是转发给其他消费者处理）
         * Global：如果为true，代表该信道上所有消费者都会被设置
         */
        //设置信道最多能存放多大的消息，最多能存放多少条未确认信息。
        consumer1.basicQos(0,3,false);
        consumer2.basicQos(0,1,false);
        //生产者线程
        new Thread(){
            @Override
            public void run() {
                try {
                    //如果声明的交换机，已经存在且类型也一致，则不会报错，也不会覆盖。
                    producer.exchangeDeclare(EXCHANGE_NAME,"direct");
                    //将queueName队列与交换机进行绑定，绑定的路由键为routingKey
                    producer.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);
                    //通过信道，与消息中间件通信。即发送消息到消息中间件。
                    Scanner in=new Scanner(System.in);
                    while(in.hasNext())
                        producer.basicPublish(EXCHANGE_NAME, ROUTING_KEY,null,in.next().getBytes(StandardCharsets.UTF_8));
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }.start();

        //消费者1线程
        ExecutorService executorService = Executors.newCachedThreadPool();
        executorService.execute(() -> {
            try {
                //消费者接受消息回调
                DeliverCallback deliverCallback=(consumerTag,publishMessage)->{
                    System.out.println("消费者1接受到"+new String(publishMessage.getBody(),StandardCharsets.UTF_8)+"，正在处理中，还未进行消息确认回应");
                    try {
                        TimeUnit.SECONDS.sleep(20);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    if(Math.ceil(Math.random()*3)==1){
                        try {
                            consumer1.close(200,"消费者1信道关闭");
                            System.out.println("********************************************");
                            System.out.println("消费者1信道关闭");
                        } catch (TimeoutException e) {
                            e.printStackTrace();
                        }
                    }else{
                        System.out.println("====消费者1接受消息回调====");
                        System.out.println(new String(publishMessage.getBody(),StandardCharsets.UTF_8));
                        System.out.println(consumerTag);
                        System.out.println("消费者1：消息处理完毕，进行消息确认回应");
                        consumer1.basicAck(publishMessage.getEnvelope().getDeliveryTag(),false);
                    }
                };
                CancelCallback cancelCallback=consumerTag->{
                    System.out.println("====消费者1取消回调====");
                    System.out.println(consumerTag);
                };
                ConsumerShutdownSignalCallback consumerShutdownSignalCallback=(consumerTag,sig)->{
                    System.out.println("====消费者1关闭回调====");
                    System.out.println(sig);
                    System.out.println(consumerTag);
                };
                //设置取消自动确认
                consumer1.basicConsume(QUEUE_NAME, false,"messageResponseMechanism",deliverCallback,cancelCallback,consumerShutdownSignalCallback);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        //消费者2线程
        executorService.execute(() -> {
            try {
                //消费者接受消息回调
                DeliverCallback deliverCallback=(consumerTag, publishMessage)->{
                    System.out.println("消费者2接受到"+new String(publishMessage.getBody(), StandardCharsets.UTF_8)+"，正在处理中，还未进行消息确认回应");
                    System.out.println("====消费者2接受消息回调====");
                    System.out.println(new String(publishMessage.getBody(), StandardCharsets.UTF_8));
                    System.out.println(consumerTag);
                    System.out.println("消费者2：消息处理完毕，进行消息确认回应");
                    /**
                     * 参数：
                     *deliveryTag – 向消息中间件（AMQP实体）发送消息状态，消息确认：  AMQP.Basic.GetOk 或者 AMQP.Basic.Deliver
                     * multiple – 是否批量确认，true代表批量确认
                     */
                    consumer2.basicAck(publishMessage.getEnvelope().getDeliveryTag(),false);
                };
                CancelCallback cancelCallback= consumerTag->{
                    System.out.println("====消费者2取消回调====");
                    System.out.println(consumerTag);
                };
                ConsumerShutdownSignalCallback consumerShutdownSignalCallback=(consumerTag, sig)->{
                    System.out.println("====消费者2关闭回调====");
                    System.out.println(sig);
                    System.out.println(consumerTag);
                };
                //设置取消自动确认
                consumer2.basicConsume(QUEUE_NAME, false,"messageResponseMechanism",deliverCallback,cancelCallback,consumerShutdownSignalCallback);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });

    }

    /**
     * 发布确认机制
     * 需要理解：
     * confirmSelect
     * ConfirmCallback和ReturnCallback、ConfirmListener区别
     * 发布失败的触发时机、发布成功的时机。发布确认的流程
     * @throws Exception
     */
    public void publishAck() throws Exception {
        int publishMessageCount=1000;
        Channel producer = RabbitMQUtil.createChannel(16);
        Channel consumer1 = RabbitMQUtil.createChannel(17);
        producer.exchangeDelete(EXCHANGE_NAME,false);
        producer.queueDelete(QUEUE_NAME,false,false);
        //如果声明的交换机，已经存在且类型也一致，则不会报错，也不会覆盖。
        producer.exchangeDeclare(EXCHANGE_NAME,"direct");
        producer.queueDeclare(QUEUE_NAME,true,false,false,null);
        //将queueName队列与交换机进行绑定，绑定的路由键为routingKey
        producer.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);
        //在此信道上开启发布确认
        producer.confirmSelect();
        //发布确认监听器
//        ConfirmListener confirmListener=new ConfirmListener() {
//            //发布被确认时，回调该方法.只要交换机接收到，就回调该方法，不管路由键存不存在。
//            @Override
//            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
//                System.out.println("#############发布成功确认###############");
//                System.out.println("deliveryTag:"+deliveryTag+"   \tmultiple:"+multiple);
//                System.out.println("取消等待！!!!!");
//
//            }
//            //发布失败时，回调该方法??????,路由键不对时，也不会触发？？？
//            @Override
//            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
//                System.out.println("#############发布失败###############");
//                System.out.println("deliveryTag:"+deliveryTag+"   \tmultiple:"+multiple);
//            }
//        };
//        producer.addConfirmListener(confirmListener);
        ConfirmCallback ackCallback=(deliveryTag,multiple)->{
            System.out.println("#############发布成功确认###############");
            System.out.println("deliveryTag:"+deliveryTag+"   \tmultiple:"+multiple);
            System.out.println("取消等待！!!!!");
        };
        ConfirmCallback nackCallback=(deliveryTag,multiple)->{
            System.out.println("#############发布失败###############");
            System.out.println("deliveryTag:"+deliveryTag+"   \tmultiple:"+multiple);
        };
        producer.addConfirmListener(ackCallback,nackCallback);
        //当队列不存在时，会回调该方法。前提是开启mandatory
        ReturnCallback returnCallback= returnMessage -> {
            System.out.println("该交换机的该路由键对应的队列不存在");
            System.out.println("#############消息回退###############");
            System.out.println(returnMessage.getReplyCode()+"  "+returnMessage.getReplyText());
        };
        producer.addReturnListener(returnCallback);
        //生产者线程
        new Thread(){
            @Override
            public void run() {
                try {
                    //通过信道，与消息中间件通信。即发送消息到消息中间件。
                    Scanner in=new Scanner(System.in);
                    while(in.hasNext()){
                        System.out.println("发布的消息的唯一标识："+producer.getNextPublishSeqNo());
                        producer.basicPublish(EXCHANGE_NAME, "ROUTING_KEY",true,null,in.next().getBytes(StandardCharsets.UTF_8));
                        System.out.println("等待发布的消息，被消息中间件确认");
//                        try {
//                            //等待，直到自上次调用waitForConfirms以来发布的所有消息都被消息中间件所ack或nack
//                            //发布一次调用一次waitForConfirms。如果返回结果为false，则会关闭channel
//                            System.out.println(producer.waitForConfirms(10000));
//                        } catch (InterruptedException e) {
//                            e.printStackTrace();
//                        } catch (TimeoutException e) {
//                            e.printStackTrace();
//                        }
                    }

                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }.start();

        new Thread(){
            @Override
            public void run() {
                try {
                    //消费者接受消息回调
                    DeliverCallback deliverCallback=(consumerTag,publishMessage)->{
                        System.out.println("消费者1接受到"+new String(publishMessage.getBody(),StandardCharsets.UTF_8)+"，正在处理中，还未进行消息确认回应");
                        if(Math.ceil(Math.random()*5)==1){
                            try {
                                consumer1.close(200,"消费者1信道关闭");
                                System.out.println("********************************************");
                                System.out.println("消费者1信道关闭");
                            } catch (TimeoutException e) {
                                e.printStackTrace();
                            }
                        }else if(Math.ceil(Math.random()*5)==2||Math.ceil(Math.random()*5)==3){
                            System.out.println("消费者进行basicNack");
                            consumer1.basicReject(publishMessage.getEnvelope().getDeliveryTag(),true);
                        }
                        else{
                            System.out.println("====消费者1接受消息回调====");
                            System.out.println(new String(publishMessage.getBody(),StandardCharsets.UTF_8));
                            System.out.println("该信道该消费者的该消息的唯一标识deliveryTag:"+publishMessage.getEnvelope().getDeliveryTag());
                            System.out.println(consumerTag);
                            System.out.println("消费者1：消息处理完毕，进行消息确认回应");
                            consumer1.basicAck(publishMessage.getEnvelope().getDeliveryTag(),false);
                        }
                    };
                    CancelCallback cancelCallback=consumerTag->{
                        System.out.println("====消费者1取消回调====");
                        System.out.println(consumerTag);
                    };
                    ConsumerShutdownSignalCallback consumerShutdownSignalCallback=(consumerTag,sig)->{
                        System.out.println("====消费者1关闭回调====");
                        System.out.println(sig);
                        System.out.println(consumerTag);
                    };
                    //设置取消自动确认
                    consumer1.basicConsume(QUEUE_NAME, false,"messageResponseMechanism",deliverCallback,cancelCallback,consumerShutdownSignalCallback);

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }.start();

    }

    /**
     * 测试死信机制
     * 需要理解：
     * 死信的来源（死信触发的时机）、死信队列、死信交换机、（队列的参数属性，以及发布的参数属性）其实是一样的（差别：是一个是该参数只针对该发布，一个是针对该队列）
     *发布参数:
     * MessageProperties类、AMQP.BasicProperties
     * 队列/发布参数：
     * x-message-ttl：消息的生存时间，发布到队列的消息在被丢弃之前可以存活多长时间(毫秒)。
     *     例如：x-message-ttl: 数值(毫秒)
     * x-expires ：队列在被自动删除之前可以使用多长时间(毫秒)。
     *     例如：x-expires: 数值(毫秒)
     * x-overflow ：设置队列溢出行为。这决定了当到达队列的最大长度时消息会发生什么。有效值为drop-head、reject-publish或reject-publish-dlx。仲裁队列类型只支持drop-head和reject-publish。
     *     例如：x-overflow: drop-head/reject-publish/reject-publish-dlx
     * x-single-active-consumer ：如果设置了，则确保每次只有一个消费者从队列中消费，并在活动消费者取消或死亡的情况下将故障转移到另一个已注册的消费者。
     *     例如：x-single-active-consumer: true/false
     * x-dead-letter-exchange ：如果消息被拒绝或过期，将向死信交换机发送该死信消息。
     *     例如：x-dead-letter-exchange: 交换机名
     * x-dead-letter-routing-key ：如果没有设置，将使用消息的原始路由键""。死信队列是要自己将死信交换机用该死信路由键绑定到死信队列的
     *     例如：x-dead-letter-routing-key: 路由键名
     * x-max-length： 该队列最大能存放多少消息，超出该长度，那么后续来的消息，将成为死信
     *     例如：x-max-length: 数值
     * x-max-length-bytes ：该队列最大能存放消息体的最大字节大小
     *     例如：x-max-length-bytes: 数值(字节)
     * x-max-priority ：队列支持的最大优先级级别数;如果没有设置，队列将不支持消息优先级。
     *     例如：x-max-priority：数值
     * x-queue-mode ：将队列设置为lazy惰性模式，在磁盘上保留尽可能多的消息以减少RAM的使用;如果没有设置，队列将保持内存缓存，以尽可能快地传递消息。
     *     例如：x-queue-mode: lazy/不设置
     * x-queue-master-locator ：将队列设置为主位置模式，确定在节点集群上声明队列主时定位该队列的规则。？？？？？？
     *     例如：
     *  交换机参数：
     *  alternate-exchange：如果到此交换器的消息不能以其他方式路由，则将它们发送到此处命名的备用交换器。
     */
    public void deadLetter() throws Exception {
        //普通交换机、普通队列
        String EXCHANGE="exchange";
        String QUEUE="queue";
        //死信交换机、死信队列、死信路由键
        String DEAD_EXCHANGE="dead_exchange";
        String DEAD_QUEUE="dead_queue";
        String DEAD_ROUTING_KEY="dead_routing_key";

        Channel producer=RabbitMQUtil.createChannel(16);
        Channel consumer=RabbitMQUtil.createChannel(17);
        Channel deadConsumer=RabbitMQUtil.createChannel(18);



        //删除之前存在的交换机，不管有没有队列使用与其交换机绑定，都将删除，如果为true，则如果有交换机与其绑定，则报错
        producer.exchangeDelete(DEAD_EXCHANGE,false);
        //同理，如果true有交换机绑定该队列，则会报错
        producer.queueDelete(DEAD_QUEUE,false,false);
        producer.exchangeDelete(EXCHANGE,false);
        producer.queueDelete(QUEUE,false,false);
        //生产者线程
        Thread producerThread=new Thread(){
            @Override
            public void run() {
                try {
                    //死信交换机的声明
                    producer.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT,false,false,false,null);
                    //死信队列的声明
                    producer.queueDeclare(DEAD_QUEUE,false,false,false,null);
                    //将死信队列以路由键为关联与死信交换机绑定
                    producer.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,DEAD_ROUTING_KEY);


                    /**
                     *方法参数：
                     * 交换机：交换机名字，如果交换机名字存在且类型一样，则不会创建交换机，但是如果类型不一样且存在，则会报错
                     * 交换机类型：常用类型direct(它会把消息，路由到那些BindingKey和RoutingKey完全匹配的队列中)、fanout（它会把所有发送到该交换器的消息，路由到所有与该交换器绑定的队列中）、topic（它也是将消息路由到BindingKey和RoutingKey相匹配的队列中，但这里的匹配规则有些不同，它可以允许使用.、*、#）
                     * 持久化：如果我们声明了一个持久的交换(交换将在服务器重启后存活)
                     * 自动删除：当没有队列、客户端使用交换机时，自动删除交换机
                     * internal：如果exchange是内部的，也就是说不能被客户端直接publish消息。
                     * 交换机参数：
                     *      * alternate-exchange：如果到此交换器的消息不能以其他方式路由，则将它们发送到此处命名的备用交换器。
                     */

                    producer.exchangeDeclare(EXCHANGE,BuiltinExchangeType.FANOUT,true,false,false ,null);
                    Map<String,Object> queueProperties=new HashMap();
                    //该队列里的所有消息的过期时间为10s，建议在发布确认那里写TTL时间，这样灵活，可以任意时间
                    //queueProperties.put("x-message-ttl",10000);
                    //队列最多可以存放10条消息
                    queueProperties.put("x-max-length",10);
                    //drop-head/reject-publish/reject-publish-dlx
                    //queueProperties.put("x-overflow",reject-publish);
                    //队列在被自动删除之前可以使用30s,一旦设置，不管有没有设置自动删除，只要空闲30s后，自动删除
                    queueProperties.put("x-expires",30000);
                    //设置该队列有死信交换机和死信队列
                    queueProperties.put("x-dead-letter-exchange",DEAD_EXCHANGE);
                    //指定死信路由键前，先指定死信交换机，死信路由键不设置，则默认为"";
                    queueProperties.put("x-dead-letter-routing-key",DEAD_ROUTING_KEY);
                    queueProperties.put("x-queue-mode", "lazy");
                    producer.queueDeclare(QUEUE,true,false,false,queueProperties);
                    //普通交换机和普通队列进行绑定,路由键不能为空的，即使fanout类型的交换机不需要路由键，但也是不能为null
                    producer.queueBind(QUEUE,EXCHANGE,"");

                    //发布参数
                    AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
                    //消息过期时间
                    builder.expiration("10000");
                    //该队列必须是优先级队列，设置消息的优先级，（消息的优先级也是必须的）
                    //builder.priority(1)
                    //插件延迟参数是没有的
                    AMQP.BasicProperties publishProperties = builder.build();

                    for(int i=1;i<=100;i++){
                        String message="message"+i;
                        producer.basicPublish(EXCHANGE,"",true,false,publishProperties,message.getBytes(StandardCharsets.UTF_8));
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        };
        producerThread.start();
        //等producerThread线程执行3秒后，主线程在开始进行执行。防止有些东西还没创建好，开始对其队列进行消费
        producerThread.join(3000);
        //普通消费者线程
        Thread consumerThread= new Thread(() -> {
            DeliverCallback deliverCallback=(consumerTag,publishMessage)->{
                System.out.println("====接受消息回调====");
                System.out.println(new String(publishMessage.getBody(),StandardCharsets.UTF_8));
                System.out.println("该信道该消费者的该消息的唯一id，deliveryTag:"+publishMessage.getEnvelope().getDeliveryTag());
                System.out.println(consumerTag);
                consumer.basicAck(publishMessage.getEnvelope().getDeliveryTag(),false);
            };
            CancelCallback cancelCallback=consumerTag->{
                System.out.println("====取消回调====");
                System.out.println(consumerTag);
            };
            ConsumerShutdownSignalCallback consumerShutdownSignalCallback=(consumerTag,sig)->{
                System.out.println("====关闭回调====");
                System.out.println(sig);
                System.out.println(consumerTag);
            };
            try {
                consumer.basicConsume(QUEUE,false,deliverCallback,cancelCallback,consumerShutdownSignalCallback);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });

        //死信线程
        new Thread(){
            @Override
            public void run() {
                DeliverCallback deliverCallback=(consumerTag,publishMessage)->{
                    System.out.println("死信信息id："+publishMessage.getEnvelope().getDeliveryTag()+"\t死信信息："+new String(publishMessage.getBody(),StandardCharsets.UTF_8));
                    deadConsumer.basicAck(publishMessage.getEnvelope().getDeliveryTag(),true);
                };
                CancelCallback cancelCallback=consumerTag->{
                    System.out.println("====死信取消回调====");
                    System.out.println(consumerTag);
                };
                ConsumerShutdownSignalCallback consumerShutdownSignalCallback=(consumerTag,sig)->{
                    System.out.println("====死信关闭回调====");
                    System.out.println(sig);
                    System.out.println(consumerTag);
                };
                try {
                    deadConsumer.basicConsume(DEAD_QUEUE,false,deliverCallback,cancelCallback,consumerShutdownSignalCallback);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }.start();
    }


    /**
     * 优先级队列测试
     */
    public void testPriority() throws Exception {
        Channel producer=RabbitMQUtil.createChannel(16);
        Channel consumer=RabbitMQUtil.createChannel(17);

        producer.exchangeDelete(EXCHANGE_NAME);
        producer.queueDelete(QUEUE_NAME);

        producer.exchangeDeclare(EXCHANGE_NAME,ExchangeTypes.DIRECT,false,false,null);
        HashMap<String, Object> properties = new HashMap<>();
        properties.put("x-max-priority",10);
        producer.queueDeclare(QUEUE_NAME,false,false,false,properties);
        producer.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);

        Thread t1=new Thread(){
            @Override
            public void run() {
                //同级别优先级
                for(int i=1;i<=10;i++){
                    //如果设置每隔1ms发一次
                    try {
                        sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    //先后顺序
                    for(int j=0;j<=10;j++){
                        String message="message"+(10*(i-1)+j);
                        //发布参数
                        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
                        //该队列必须是优先级队列，设置消息的优先级，（消息的优先级也是必须的）
                        builder.priority(i);
                        AMQP.BasicProperties publishProperties = builder.build();
                        try {
                            //如果设置每隔1ms发一次，会发现结果是按发送顺序处理
//                            sleep(1);
                            //如果直接一次性发很多条，发现一开始按优先级的，但后面却不按优先级
                            producer.basicPublish(EXCHANGE_NAME,ROUTING_KEY,true,false,publishProperties,message.getBytes(StandardCharsets.UTF_8));
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
                System.out.println("发送完了");
            }
        };
        t1.start();

//        //如果在队列中堆积,并给足时间让其排序,则出来的结果一定是从大到小,先到先处理的结果
//        t1.join();
//        TimeUnit.SECONDS.sleep(5);

        System.out.println("开始接收");
        new Thread(){
            @Override
            public void run() {
                DeliverCallback deliverCallback = (consumerTag, publishMessage) -> {
//                    try {
//                        sleep(1);
//                    } catch (InterruptedException e) {
//                        e.printStackTrace();
//                    }
                    System.out.printf("消息: %s\n",new String(publishMessage.getBody(), StandardCharsets.UTF_8));
                    consumer.basicAck(publishMessage.getEnvelope().getDeliveryTag(),false);
                };
                CancelCallback cancelCallback = consumerTag -> {
                    System.out.println("====取消回调====");
                    System.out.println(consumerTag);
                };
                ConsumerShutdownSignalCallback consumerShutdownSignalCallback = (consumerTag, sig) -> {
                    System.out.println("====关闭回调====");
                    System.out.println(sig);
                    System.out.println(consumerTag);
                };
                try {
                    consumer.basicConsume(QUEUE_NAME, false, "test", deliverCallback, cancelCallback, consumerShutdownSignalCallback);
                } catch (IOException e) {
                    e.printStackTrace();
                }

            }
        }.start();

    }

    /**
     * 持久化
     * 需要理解：
     * 交换机持久化、队列持久化、信息持久化。
     * 发布确认模式来保证消息不丢失
     * 消息应答来保证消息不丢失
     */
    public void dataPersistence() throws Exception {

    }





}
